package org.example.catalog;

import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.TableResult;

public class FlinkHiveCatalog {
    public static void main(String[] args) throws Exception {
        // 使用批处理模式创建TableEnvironment
        EnvironmentSettings settings = EnvironmentSettings
                .newInstance()
                .inBatchMode() // 设置为批处理模式
                .build();
        TableEnvironment tableEnv = TableEnvironment.create(settings);

        // 创建Paimon catalog
        String createCatalogSQL = "CREATE CATALOG paimon WITH (\n" +
                "    'type' = 'paimon',\n" +
                "    'metastore' = 'hive',\n" +
                "    'uri' = 'thrift://hadoop1:9083'" +
                ");";
        System.out.println("执行SQL: \n" + createCatalogSQL);
        tableEnv.executeSql(createCatalogSQL);
        
        // 创建Paimon数据库
        String createDatabaseSQL = "CREATE DATABASE IF NOT EXISTS `paimon`.`paimon`;";
        System.out.println("执行SQL: \n" + createDatabaseSQL);
        tableEnv.executeSql(createDatabaseSQL);

        // 删除已存在的用户表（如果存在）
        String dropTableSQL = "DROP TABLE IF EXISTS `paimon`.`paimon`.`t_user`;";
        System.out.println("执行SQL: \n" + dropTableSQL);
        tableEnv.executeSql(dropTableSQL);

        // 创建带主键的用户表
        String createTableSQL = "CREATE TABLE IF NOT EXISTS `paimon`.`paimon`.`t_user` (\n" +
                "    user_id BIGINT PRIMARY KEY NOT ENFORCED COMMENT '用户ID（主键）',\n" +
                "    user_name STRING COMMENT '用户名'\n" +
                ");";
        System.out.println("执行SQL: \n" + createTableSQL);
        tableEnv.executeSql(createTableSQL);

        // 直接插入多条数据，确保数据被写入
        String insertSQL = "INSERT INTO `paimon`.`paimon`.`t_user` VALUES " +
                "(1, '张三'), " +
                "(2, '李四'), " +
                "(3, '王五');";
        System.out.println("执行SQL: \n" + insertSQL);
        TableResult insertResult = tableEnv.executeSql(insertSQL);

        // 等待插入操作完成
        insertResult.await();
        
        System.out.println("数据插入完成，准备查询...");

        // 查询数据并打印
        String selectSQL = "SELECT * FROM `paimon`.`paimon`.`t_user`;";
        System.out.println("执行SQL: \n" + selectSQL);
        tableEnv.executeSql(selectSQL).print();
        
        // 确保所有操作都已经完成
        System.out.println("查询执行完毕");
    }
}